package com.gin.monitor.agg.warn;

import com.gin.monitor.agg.utils.JobEnvUtils;
import com.gin.monitor.agg.vo.PoliceAction;
import com.gin.monitor.agg.vo.TrafficInfo;
import com.gin.monitor.agg.vo.ViolationInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;


/**
 * 违法车辆和交警出警分析
 * 第一种，当前的违法车辆（在5分钟内）如果已经出警了。（最后输出道主流中做删除处理）。
 * 第二种，当前违法车辆（在5分钟后）交警没有出警（发出出警的提示，在侧流中发出）。
 * 第三种，有交警的出警记录，但是不是由监控平台报的警。
 * 需要两种数据流：
 * 1、系统的实时违法车辆的数据流
 * 2、交警实时出警记录数据
 * <p>
 * <p>
 * 测试时注意: 使用当前时间戳
 * nc -lk 9999
京L22189,套牌,1620637884000
京L22188,套牌,1620637885000
京L22187,套牌,1620637886000
京L22187,套牌,1620637996000
 * <p>
 * nc -lk 8888
007,京L22187,出警,1620637885000
008,京L22188,处理完成,1620637886000
 *
 * @author gin
 * @date 2021/4/30
 */
public class ViolationCarAndPoliceActionAnalysis2 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = JobEnvUtils.initEnv(args);

        //第二种，当前违法车辆（在5分钟后）交警没有出警（发出出警的提示，在侧流中发出）。
        // nc -lk 9999
        SingleOutputStreamOperator<ViolationInfo> stream1 = env.socketTextStream("node01", 9999)
                .map(new RichMapFunction<String, ViolationInfo>() {
                    @Override
                    public ViolationInfo map(String line) {
                        String[] s = line.split(",");
                        return new ViolationInfo(s[0], s[1], Long.parseLong(s[2]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<ViolationInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((event, timestamp) -> event.createTime));

        // nc -lk 8888
        SingleOutputStreamOperator<PoliceAction> stream2 = env.socketTextStream("node01", 8888)
                .map(new RichMapFunction<String, PoliceAction>() {
                    @Override
                    public PoliceAction map(String line) {
                        String[] s = line.split(",");
                        return new PoliceAction(s[0], s[1], s[2], Long.parseLong(s[3]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<PoliceAction>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((event, timestamp) -> event.actionTime));

        //是否违法车辆
        OutputTag<PoliceAction> tag1 = new OutputTag<PoliceAction>("No Violaction Car!"){};
        //是否已出警处理
        OutputTag<ViolationInfo> tag2 = new OutputTag<ViolationInfo>("No PoliceAction!"){};

        SingleOutputStreamOperator<String> mainStream = stream1.keyBy((KeySelector<ViolationInfo, String>) violationInfo -> violationInfo.car)
                .connect(stream2.keyBy((KeySelector<PoliceAction, String>) policeAction -> policeAction.car))
                .process(new KeyedCoProcessFunction<String, ViolationInfo, PoliceAction, String>() {

                    //需要两个状态，分别保存违法数据，出警记录
                    ValueState<ViolationInfo> vState;
                    ValueState<PoliceAction> pState;

                    @Override
                    public void open(Configuration parameters) {
                        //状态中保存第一辆出现的信息对象
                        vState = getRuntimeContext().getState(new ValueStateDescriptor<>("v", ViolationInfo.class));
                        pState = getRuntimeContext().getState(new ValueStateDescriptor<>("p", PoliceAction.class));
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        //触发器触发的函数
                        ViolationInfo info = vState.value();
                        PoliceAction action = pState.value();
                        //表示有出警记录，但是没有匹配的违法车辆
                        if (info == null && action != null) {
                            ctx.output(tag1, action);
                        }
                        //有违法车辆信息，但是5分钟内还没有出警记录
                        if (action == null && info != null) {
                            ctx.output(tag2, info);
                        }
                        //清空状态
                        pState.clear();
                        vState.clear();
                    }

                    @Override
                    public void processElement1(ViolationInfo value, Context ctx, Collector<String> out) throws Exception {
                        PoliceAction policeAction = pState.value();
                        if (policeAction == null) {//可能出警的数据还没有读到，或者该违法处理还没有交警出警
                            ctx.timerService().registerEventTimeTimer(value.createTime + 5000); //5秒后触发提示
                            vState.update(value);
                        } else { //已经有一条与之对应的出警记录,可以关联
                            out.collect("该违法车辆" +
                                    value.car + "，违法时间" +
                                    value.createTime + ",已经有交警出警了，警号为:" +
                                    policeAction.policeId + ",出警的状态是：" +
                                    policeAction.actionStatus + ",出警的时间:" + policeAction.actionTime);
                            vState.clear();
                            pState.clear();
                        }
                    }

                    @Override
                    public void processElement2(PoliceAction value, Context ctx, Collector<String> out) throws Exception {
                        ViolationInfo info = vState.value();
                        if (info == null) { //出警记录没有找到对应的违法车辆信息
                            ctx.timerService().registerEventTimeTimer(value.actionTime + 5000);
                            pState.update(value);
                        } else {//已经有一条与之对应的出警记录,可以关联
                            out.collect("该违法车辆" +
                                    info.car + "，违法时间" +
                                    info.createTime + ",已经有交警出警了，警号为:" +
                                    value.policeId + ",出警的状态是：" +
                                    value.actionStatus + ",出警的时间:" + value.actionTime);
                            vState.clear();
                            pState.clear();
                        }
                    }
                });

        mainStream.print();
        mainStream.getSideOutput(tag1).print("没有对应的违法车辆信息");
        mainStream.getSideOutput(tag2).print("该违法车辆在5分钟内没有交警出警");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
